/**
 * 
 */
package org.pump.client.main;

import java.net.InetSocketAddress;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.mina.core.future.ConnectFuture;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.DefaultSocketSessionConfig;
import org.apache.mina.transport.socket.nio.NioSocketConnector;
import org.pump.client.codec.EventData;
import org.pump.client.pool.PumpPool;
import org.pump.client.util.MyConstant;
import org.pump.client.util.PumpUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author liudong03
 * 
 */
public class MyClient {
	private static final Logger mylog = LoggerFactory.getLogger(MyClient.class);
	
	MyProperty conf;
	MysqlEncoder encoder;
	MysqlDecoder decoder;
	ClientHandler handler;
	PumpPool eventpool;

	NioSocketConnector connector;
	ConnectFuture future;

	public boolean init(MyProperty conf,PumpPool eventPool) {
		if (conf == null) {
			mylog.error("the configuration properties is null");
			return false;
		}
	
		this.conf = conf;
		this.eventpool = eventPool;
		this.decoder = new MysqlDecoder();
		this.encoder = new MysqlEncoder();
		this.handler = new ClientHandler();
		this.handler.init(conf, eventPool,this);
		this.connector = new NioSocketConnector();
		
		
		this.connector.setConnectTimeoutMillis(this.conf.conTimeOut);
		this.connector.getFilterChain().addLast("codec",
				new ProtocolCodecFilter(this.encoder, this.decoder));
		// connector.getFilterChain().addLast("logger", new LoggingFilter());
		this.connector.setHandler(handler);

		DefaultSocketSessionConfig sessionconfig = (DefaultSocketSessionConfig) connector
				.getSessionConfig();
		sessionconfig.setReadBufferSize(this.conf.readBufSize);
		sessionconfig.setSendBufferSize(this.conf.writeBufSize);
		sessionconfig.setReuseAddress(true);
		sessionconfig.setTcpNoDelay(true);
		sessionconfig.setKeepAlive(true);
		sessionconfig.setSoLinger(-1);
		sessionconfig.setWriteTimeout(5000);
		//sessionconfig.setMaxReadBufferSize(16 * 1024 * 1024);
		
		return true;
	}

	public int start() {
		
		this.future = connector.connect(new InetSocketAddress(
				this.conf.hostname, this.conf.hostport));
		this.future.awaitUninterruptibly(this.conf.conTimeOut,TimeUnit.SECONDS);
		if (!this.future.isConnected()) {
			mylog.error("connect to " + this.conf.hostname + ":"
					+ Integer.toString(this.conf.hostport)
					+ " fail");
			this.connector.dispose();
			return MyConstant.CONNECT_FAIL;
		}
		mylog.info("connect to " + this.conf.hostname + ":"
				+ Integer.toString(this.conf.hostport) + " succ");
		
		
		return MyConstant.SUCC;
	}

	public void stop() {
		this.eventpool.commit();
		IoSession cursession = this.future.getSession();
		if(PumpUtil.activeCloseSession(cursession))
		{
			this.connector.dispose();
		}else
		{
			mylog.error("active close session fail");
			return;
		}
	}
}
